package com.study.flink.java.day04_async.async_io.http;

import com.study.flink.java.day04_async.ActivityBean;
import com.study.flink.java.utils.FlinkUtilsDemo;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import java.util.concurrent.TimeUnit;

/**
 * 异步的从外部调用http取数
 * 最后sink打印
 */
public class QueryActivityAsyncBalanceMain {

    public static void main(String[] args) throws Exception {
        DataStream<String> lines = FlinkUtilsDemo.createKafkaStream(args, new SimpleStringSchema());
        //SingleOutputStreamOperator<ActivityBean> beans = lines.map(new BalanceAsyncToActivityBeanFunction());
        // 容量数不能大于http异步请求数量
        SingleOutputStreamOperator<ActivityBean> result = AsyncDataStream.unorderedWait(
                lines, new BalanceAsyncToActivityBeanFunction(), 5000, TimeUnit.MILLISECONDS, 10);
        result.print();
        FlinkUtilsDemo.getEnv().execute("QueryActivityAsyncBalance");
    }


}
